/*
 * Copyright (c) 2019.  drakelee.base All rights reserved.
 */

package com.base.components.task.service;

import com.base.components.task.controller.TaskBroadcastController;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.base.components.common.service.ThreadPoolService;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.NameValuePair;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;

/**
 * 作业服务，发现服务并广播
 *
 * @author <a href="drakelee1221@gmail.com">LiGeng</a>
 * @version 1.0.0, 2017-08-03 15:03
 *
 */
@Service
public class ServerClientService {

  private static final Logger logger = LoggerFactory.getLogger(ServerClientService.class);

  @Autowired
  private ThreadPoolService threadPoolService;

  @Autowired
  private DiscoveryClient discoveryClient;

  @Value("${module-task}")
  private String taskServerName;

  public List<ServiceInstance> findAllServers() {
    return discoveryClient.getInstances(taskServerName);
  }

  /**
   * 广播调度作业
   * @param type 广播类别
   * @param async 是否异步
   * @return 异步返回 -1，同步返回成功广播的服务器总数
   */
  public int sendBroadcast(TaskBroadcastType type, boolean async) {
    List<ServiceInstance> servers = findAllServers();
    if (servers.size() > 0) {
      if(async){
        asyncBroadcast(servers, Collections.emptyList(), type, null);
        return -1;
      }
      return broadcast(servers, Collections.emptyList(), type, null);
    } else {
      return 0;
    }
  }
  /**
   * 广播调度作业
   * @param type 广播类别
   * @param async 是否异步
   * @param extendParam 拓展参数
   * @return 异步返回 -1，同步返回成功广播的服务器总数
   */
  public int sendBroadcast(TaskBroadcastType type, boolean async, Map<String,String> extendParam) {
    List<ServiceInstance> servers = findAllServers();
    if (servers.size() > 0) {
      if(async){
        asyncBroadcast(servers, Collections.emptyList(), type, extendParam);
        return -1;
      }
      return broadcast(servers, Collections.emptyList(), type, extendParam);
    } else {
      return 0;
    }
  }

  /**
   * 广播调度作业
   * @param fireTaskId 作业id
   * @param type 广播类别
   * @param async 是否异步
   * @return 异步返回 -1，同步返回成功广播的服务器总数
   */
  public int sendBroadcast(String fireTaskId, TaskBroadcastType type, boolean async) {
    List<ServiceInstance> servers = findAllServers();
    if (servers.size() > 0) {
      if(async){
        asyncBroadcast(servers, Lists.newArrayList(fireTaskId), type, null);
        return -1;
      }
      return broadcast(servers, Lists.newArrayList(fireTaskId), type, null);
    } else {
      return 0;
    }
  }

  /**
   * 广播调度作业
   * @param fireTaskId 作业id
   * @param type 广播类别
   * @param async 是否异步
   * @param extendParam 拓展参数
   * @return 异步返回 -1，同步返回成功广播的服务器总数
   */
  public int sendBroadcast(String fireTaskId, TaskBroadcastType type, boolean async, Map<String,String> extendParam) {
    List<ServiceInstance> servers = findAllServers();
    if (servers.size() > 0) {
      if(async){
        asyncBroadcast(servers, Lists.newArrayList(fireTaskId), type, extendParam);
        return -1;
      }
      return broadcast(servers, Lists.newArrayList(fireTaskId), type, extendParam);
    } else {
      return 0;
    }
  }

  /**
   * 广播调度作业
   * @param fireTaskIds 作业id集合
   * @param type 广播类别
   * @param async 是否异步
   * @return 异步返回 -1，同步返回成功广播的服务器总数
   */
  public int sendBroadcast(List<String> fireTaskIds, TaskBroadcastType type, boolean async) {
    List<ServiceInstance> servers = findAllServers();
    if (servers.size() > 0) {
      if(async){
        asyncBroadcast(servers, fireTaskIds, type, null);
        return -1;
      }else{
        return broadcast(servers, fireTaskIds, type, null);
      }
    }
    return 0;
  }

  /**
   * 广播调度作业
   * @param fireTaskIds 作业id集合
   * @param type 广播类别
   * @param async 是否异步
   * @param extendParam 拓展参数
   * @return 异步返回 -1，同步返回成功广播的服务器总数
   */
  public int sendBroadcast(List<String> fireTaskIds, TaskBroadcastType type, boolean async, Map<String,String> extendParam) {
    List<ServiceInstance> servers = findAllServers();
    if (servers.size() > 0) {
      if(async){
        asyncBroadcast(servers, fireTaskIds, type, extendParam);
        return -1;
      }else{
        return broadcast(servers, fireTaskIds, type, extendParam);
      }
    }
    return 0;
  }

  /** 主线程同步-子线程异步 广播 */
  private int broadcast(List<ServiceInstance> servers, List<String> fireTaskIds, TaskBroadcastType type, Map<String,String> extendParam) {
    final Map<String, String> params = Maps.newHashMap();
    if(extendParam != null && !extendParam.isEmpty()){
      params.putAll(extendParam);
    }
    params.put(TaskBroadcastController.BROADCAST_TYPE_KEY, type.toString());
    String ids = StringUtils.join(fireTaskIds, TaskBroadcastController.TASK_FIRE_IDS_SEPARATOR_CHARS);
    if (null != fireTaskIds && !fireTaskIds.isEmpty()) {
      params.put(TaskBroadcastController.TASK_FIRE_IDS_KEY, ids);
    }
    logger.info("-----同步调度任务广播开始：{}{}", type.getDesc(),
                (null != fireTaskIds && !fireTaskIds.isEmpty()?
                 "(fireTaskIds:" + fireTaskIds + ")" : ""));
    final CountDownLatch latch = new CountDownLatch(servers.size());
    List<Future<String>> futures = Lists.newArrayList();
    for (final ServiceInstance server : servers) {
      futures.add(threadPoolService.call(new Callable<String>() {
        @Override
        public String call() throws Exception {
          try {
            String result = doPost(server.getUri() + TaskBroadcastController.BROADCAST_URL, params);
            return server.getUri() + " => " + result;
          } finally {
            latch.countDown();
          }
        }
      }));
    }
    try {
      latch.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    int successCount = 0;
    for (Future<String> fu : futures) {
      try {
        String result = fu.get();
        if (!Strings.isNullOrEmpty(result) && result
            .contains(TaskBroadcastController.BroadcastReturn.SUCCESS.toString())) {
          successCount++;
          logger.info(result);
        } else {
          logger.error(result);
        }
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
    logger.info("-----同步调度任务广播结束, 成功 : {}, {}{}", successCount, type.getDesc(),
                (null != fireTaskIds && !fireTaskIds.isEmpty()? "(fireTaskId:" + fireTaskIds + ")" : ""));
    return successCount;
  }

  /** 异步广播 */
  private void asyncBroadcast(List<ServiceInstance> servers, List<String> fireTaskIds, TaskBroadcastType type, Map<String,String> extendParam){
    final Map<String,String> params = Maps.newHashMap();
    if(extendParam != null && !extendParam.isEmpty()){
      params.putAll(extendParam);
    }
    params.put(TaskBroadcastController.BROADCAST_TYPE_KEY, type.toString());
    String ids = StringUtils.join(fireTaskIds, TaskBroadcastController.TASK_FIRE_IDS_SEPARATOR_CHARS);
    if (null != fireTaskIds && !fireTaskIds.isEmpty()) {
      params.put(TaskBroadcastController.TASK_FIRE_IDS_KEY, ids);
    }
    logger.info("-----异步调度任务广播：{}{}", type.getDesc(),
                (null != fireTaskIds && !fireTaskIds.isEmpty()?"(fireTaskId:"+fireTaskIds+")":""));
    for(final ServiceInstance server : servers){
      threadPoolService.run(new Runnable() {
        @Override
        public void run() {
          String result = doPost(server.getUri() + TaskBroadcastController.BROADCAST_URL, params);
          logger.info(server.getUri() + " => " + result);
        }
      });
    }
  }



  private String doPost(String url, Map<String, String> params) {
    String respStr = null;
    try (CloseableHttpClient hc = HttpClients.createDefault()) {
      List<NameValuePair> pairs = null;
      if (params != null && !params.isEmpty()) {
        pairs = Lists.newArrayListWithCapacity(params.size());
        for (Map.Entry<String, String> entry : params.entrySet()) {
          String value = entry.getValue();
          if (value != null) {
            pairs.add(new BasicNameValuePair(entry.getKey(), value));
          }
        }
      }
      HttpPost httpPost = new HttpPost(url);
      httpPost.addHeader(TaskBroadcastController.BROADCAST_AUTH_HEADER_KEY, TaskBroadcastController.BROADCAST_AUTH_HEADER_VALUE);
      if (pairs != null && pairs.size() > 0) {
        httpPost.setEntity(new UrlEncodedFormEntity(pairs, "UTF-8"));
      }
      HttpResponse response = hc.execute(httpPost);
      HttpEntity entity = response.getEntity();
      if (entity != null) {
        respStr = EntityUtils.toString(entity, "UTF-8");
      }
      EntityUtils.consume(entity);
    } catch (IOException e) {
      e.printStackTrace();
      respStr = ">>> 调度任务广播服务节点失败 => " + url + params;
    }
    return respStr;
  }




}
